package com.vodafone.mapreduce.divide;

import java.util.ArrayList;
import java.util.List;

import javax.jms.MessageProducer;
import javax.jms.Session;

/**
 * TODO: Insert your comments here!
 */
public class WordFrequencyDivideFunction extends DivideFunction<String, String> {

    public static final int DIVIDE_AFTER_LINES = 10;

    private final int divideAfterLines;

    public WordFrequencyDivideFunction(String k, String v) {
        super(k, v);
        this.divideAfterLines = DIVIDE_AFTER_LINES;
    }

    public WordFrequencyDivideFunction(String k, String v, int l) {
        super(k, v);
        this.divideAfterLines = l;
    }

    public WordFrequencyDivideFunction(String k, String v, int l, List<Session> jmsSessions, List<MessageProducer> messageProducers) {
        super(k, v, jmsSessions, messageProducers);
        this.divideAfterLines = l;
    }

    @Override
    protected void doDivide() {
        String[] chunks = org.apache.commons.lang.StringUtils.splitPreserveAllTokens(v, '\n');
        List<String> v2 = new ArrayList<String>();
        for (int i = 0; i < chunks.length; i++) {
            if (i == chunks.length - 1 && chunks[i].length() == 0) {
                // Skip last empty line.
            } else {
                v2.add(chunks[i]);
                this.sendChunk(k, chunks[i]);
            }
        }
        intermediate.put(k, v2);
    }

    /**
     * @return 1 - we currently don't pay attention to this, every line will be a chunk.
     */
    public int getDivideAfterLines() {
        return 1;
    }

}
